package com.butterfly.hotkey.worker.framework.disruptor;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.SystemClock;
import com.butterfly.hotkey.core.cache.ICacheManager;
import com.butterfly.hotkey.core.constant.Constants;
import com.butterfly.hotkey.core.model.HotKeyModel;
import com.butterfly.hotkey.core.rule.ClientKeyRule;
import com.butterfly.hotkey.core.util.KeyRuleUtil;
import com.butterfly.hotkey.common.disruptor.annatation.EventProcessor;
import com.butterfly.hotkey.common.disruptor.event.QueueEvent;
import com.butterfly.hotkey.common.disruptor.processor.IQueueEventProcessor;
import com.butterfly.hotkey.worker.constants.TopicConst;
import com.butterfly.hotkey.worker.framework.component.WorkerContainer;
import com.butterfly.hotkey.worker.framework.netty.pusher.IPusher;
import com.butterfly.hotkey.worker.holder.KeyCacheHolder;
import com.butterfly.hotkey.worker.util.WorkerUtil;
import com.butterfly.hotkey.worker.window.SlidingWindow;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;

/**
 * 处理从client端发过来的key
 *
 * @author butterfly
 * @version 1.0
 */
@Component
@EventProcessor(topic = TopicConst.TOPIC_NEW_KEY)
public class NewKeyProcessor implements IQueueEventProcessor {

    private static final Logger log = LoggerFactory.getLogger(Constants.WORKER);

    @Resource
    private ICacheManager hotKeyCache;

    @Resource
    private WorkerContainer workerContainer;

    @Resource
    private List<IPusher> pushers;

    @Override
    public void onEvent(QueueEvent event) {
        if (event == null) {
            return;
        }

        try {
            HotKeyModel data = (HotKeyModel) event.getData();
            handleReceiveKey(data);
        } catch (Exception e) {
            log.error("handle NewKeyProcessor error,{}", event, e);
        }
    }

    private void handleReceiveKey(HotKeyModel hotKeyModel) {
        String cacheKey = WorkerUtil.buildCacheKey(hotKeyModel);
        // 最近刚hot过的key不再进行重复计算
        boolean isHotKey = hotKeyCache.contains(cacheKey);
        if (isHotKey) {
            return;
        }

        // 获取当前key的滑动窗口,一个appName一个Cache
        String appName = hotKeyModel.getAppName();
        long accessCount = hotKeyModel.getCount().sum();

        // 对每个Key的滑动窗口进行
        ICacheManager windowCache = KeyCacheHolder.getWindowCacheManager(appName);
        SlidingWindow slidingWindow = getSlidingWindow(windowCache, hotKeyModel);
        if (slidingWindow == null) {
            return;
        }

        boolean isHot = slidingWindow.increment(accessCount);
        if (!isHot) {
            // 如果没hot，重新put，续期缓存
            windowCache.put(cacheKey, slidingWindow);
            return;
        }

        // key 已变热,准备推送client、dashboard
        handleKeyHot(cacheKey, windowCache, hotKeyModel);
    }

    private SlidingWindow getSlidingWindow(ICacheManager windowCache, HotKeyModel hotKeyModel) {
        String appName = hotKeyModel.getAppName();
        String key = hotKeyModel.getKey();

        Map<String, List<ClientKeyRule>> appRuleMap = workerContainer.getAppRuleMap();
        List<ClientKeyRule> clientKeyRules = appRuleMap.get(appName);
        if (CollUtil.isEmpty(clientKeyRules)) {
            log.warn("app:{} rule is empty", appName);
            return null;
        }
        ClientKeyRule keyRule = KeyRuleUtil.findKeyRule(key, clientKeyRules);
        if (keyRule == null) {
            log.warn("app:{} key:{} rule is empty", appName, key);
            return null;
        }
        long intervalInMs = keyRule.getIntervalInSecs() * 1000L;
        int threshold = keyRule.getThreshold();
        return windowCache.putIfAbsent(appName, new SlidingWindow(intervalInMs, threshold));
    }

    private void handleKeyHot(String cacheKey, ICacheManager windowCache, HotKeyModel hotKeyModel) {
        // 暂存hotkey，避免推送给client端重复的hotkey
        hotKeyCache.put(cacheKey, new Object());

        // 删除滑动窗口缓存，供下次检测使用
        windowCache.invalidate(cacheKey);

        // 推送给client、dashboard
        hotKeyModel.setCreateTime(SystemClock.now());
        for (IPusher pusher : pushers) {
            pusher.push(hotKeyModel);
        }
    }
}
